---
title: "05-Python 实战 psycopg 与生态"
created: 2026-08-31
tags:
  - 项目筑基
---

# Python 实战 psycopg 与生态

> PostgreSQL 系统讲解第五篇：Python 侧怎么用 PG。对齐 SQLite 篇的"三板斧"风格——连接、参数化、事务、连接池，再加备份运维和 pgvector 衔接（呼应 03-向量数据库组）。

## psycopg 3：连接与基本操作

```bash
pip install "psycopg[binary]"
```

```python
import psycopg

# 连接字符串与 psql 一致；上下文管理器自动关闭
with psycopg.connect("postgresql://localhost/mydb") as conn:
    with conn.cursor() as cur:
        # 参数化占位符是 %s——与 sqlite3 的 ? 不同，别混
        cur.execute("SELECT id, name FROM users WHERE age > %s", (18,))
        for id_, name in cur.fetchall():
            print(id, name)

        # 写操作：conn（非 autocommit 时）提交即生效，异常自动回滚
        cur.execute(
            "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id",
            ("张三", "z@ex.com"),
        )
        new_id = cur.fetchone()[0]      # RETURNING 直接拿回自增 id
    conn.commit()                       # with conn 退出也会提交
```

- **占位符是 `%s`**（不是 `?`），参数走元组——永远参数化，别拼字符串（和 SQLite/MySQL 同一条铁律）
- `psycopg` 与 `psycopg2`：新项目直接用 psycopg 3（API 更现代，原生支持管道/异步）；老代码里 psycopg2 仍常见
- 字典行：`cur = conn.cursor(row_factory=psycopg.rows.dict_row)`，取值 `row["name"]`

## 事务与连接池

```python
# 显式事务
with psycopg.connect(uri) as conn:
    try:
        with conn.transaction():
            cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = %s", (1,))
            cur.execute("UPDATE accounts SET balance = balance + 100 WHERE id = %s", (2,))
        # with transaction 正常退出=提交，异常=回滚（支持嵌套=保存点）
    except psycopg.errors.SerializationFailure:
        ...   # 串行化冲突：重试整个事务（04 篇讲过 SSI 会主动回滚）

# 服务端/长驻程序用连接池
from psycopg_pool import ConnectionPool
pool = ConnectionPool(uri, min_size=2, max_size=10)
with pool.connection() as conn:       # 用完自动归还
    ...
```

对照记忆：sqlite3 的 `with conn` ≈ psycopg 的 `with conn.transaction()`；进程内脚本用短连接，服务里必须连接池（PG 每个连接是一个后端进程，建连很重）。

## 备份与常用运维

```bash
pg_dump -U user -d mydb -Fc -f mydb.dump        # 自定义格式压缩备份（推荐 -Fc）
pg_restore -U user -d mydb_new --clean mydb     # 恢复
pg_dump mydb | psql otherdb                     # 小库快速复制
```

```sql
SELECT pg_size_pretty(pg_database_size(current_database()));   -- 库大小
SELECT relname, n_dead_tup FROM pg_stat_user_tables ORDER BY n_dead_tup DESC LIMIT 5;  -- 死元组大户
SELECT pid, state, query, now()-query_start AS running
FROM pg_stat_activity WHERE state != 'idle';                   -- 正在跑的查询（找长事务）
```

## 生态衔接：pgvector 与本组向量篇

```sql
CREATE EXTENSION vector;                                        -- 安装扩展后
CREATE TABLE docs (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, content text, emb vector(1024));
CREATE INDEX ON docs USING hnsw (emb vector_cosine_ops);        -- HNSW 索引（03-向量数据库篇讲过的机制在这里落地）

SELECT content FROM docs ORDER BY emb <=> '[0.1, 0.2, ...]' LIMIT 5;   -- 余弦距离最近邻
```

- `<->` L2 距离、`<#>` 内积、`<=>` 余弦——与"01-相似度与向量索引"篇的三度量一一对应
- 这就是"01 篇选型直觉"里**已有 PG 就先试 pgvector**的原因：业务数据和向量同库，JOIN 和事务都是现成的
- 小规模起步完全够用；亿级再考虑专用向量库（概览篇的选型直觉在这里闭环）

> 💡 系列总结：01 对照入门 → 02 类型与表设计 → 03 查询进阶 → 04 索引与 MVCC → 本篇 Python 落地。下一步实操：本地装一个 PG，把笔记工具的某个功能（比如笔记标签检索）用 jsonb/数组 + GIN 重做一遍——有真实场景，知识才挂得住。

---
⬅️ [[04-PostgreSQL 索引与 MVCC|04-PostgreSQL 索引与 MVCC]] 🏠 [[00-数据库|00-数据库]] ➡️ [[00-NoSQL 数据库总览|00-NoSQL 数据库总览]]